package consumer

import (
	"context"
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/sirupsen/logrus"
	event2 "go_knowledge/mq/kafka/event"
	"time"
)

type defaultClient struct {
	addrs  []string
	group  string
	topic  string
	logger *logrus.Logger

	cg     *sarama.ConsumerGroup
	config *sarama.Config
}

func NewDefaultConsumerClient(addrs []string, topic, group string) *defaultClient {
	config := sarama.NewConfig()
	config.Version = sarama.V0_10_2_0
	//config.Version = sarama.V2_1_0_0
	//config.Consumer.Return.Errors = false

	client := &defaultClient{
		topic:  topic,
		addrs:  addrs,
		group:  group,
		config: config,
	}

	return client
}

func (dcc *defaultClient) SetLogger(logger *logrus.Logger) {
	dcc.logger = logger

}

func (dcc *defaultClient) SetTopic(topic string) {
	if topic == "" {
		dcc.topic = topic
	}
}

func (dcc *defaultClient) start() {

	// 启动 consumerGroup
	cg, err := sarama.NewConsumerGroup(dcc.addrs, dcc.group, dcc.config)
	if err != nil {
		dcc.logger.Errorf("creating consumer group client failed: %v", err)
		return
	}

	dcc.logger.Debugf("starting a new Sarama consumer, group[%s]", dcc.group)
	dcc.cg = &cg
}

func (dcc *defaultClient) Handle(h event2.Handler) {

	// 如果未设置 logger，默认初始化一个
	if nil == dcc.logger {
		l := logrus.New()
		l.SetLevel(logrus.DebugLevel)
		l.SetFormatter(&logrus.TextFormatter{
			TimestampFormat: "2006-01-02 15:04:05.000",
		})
		dcc.logger = l
		fmt.Println("not set logger, use default &log.Logger{}")
	}

	// 启动 kafka ConsumerGroup
	dcc.start()

	// 生成 handler
	handler := NewDefaultConsumer(h)
	handler.SetLogger(dcc.logger)

	ctx, _ := context.WithCancel(context.Background())

	go func() {
		dcc.logger.Debugf("topic[%s] consumer handle start", dcc.topic)
		for {
			if err := (*dcc.cg).Consume(ctx, []string{dcc.topic}, handler); err != nil {
				dcc.logger.Errorf("topic[%s] consume failed: %s", dcc.topic, err)
				time.Sleep(3 * time.Second)
			}

			// check if context was cancelled, signaling that the consumer should stop
			if ctx.Err() != nil {
				dcc.logger.Errorf("topic[%s] consume context happen error, break: %s", dcc.topic, ctx.Err())
				return
			}
		}
	}()
}

func (dcc *defaultClient) Close() error {
	if nil != dcc.cg {
		return (*dcc.cg).Close()
	}

	return nil
}
